 1.Kafka高级特性解析之稳定性中的事务
   
   一、事务场景
   1).如producer发的多条消息组成一个事务这些消息需要对consumer同时可见或者同时不可见 。
   2).producer可能会给多个topic，多个partition发消息，这些消息也需要能放在一个事务里面，这就形成了一
个典型的分布式事务。
   3).kafka的应用场景经常是应用先消费一个topic，然后做处理再发到另一个topic，这个consume-transformproduce
过程需要放到一个事务里面，比如在消息处理或者发送的过程中如果失败了，消费偏移量也不能提交。
   4).producer或者producer所在的应用可能会挂掉，新的producer启动以后需要知道怎么处理之前未完成的事
务 。
   5).在一个原子操作中，根据包含的操作类型，可以分为三种情况，前两种情况是事务引入的场景，最后一种没用。
          1).只有Producer生产消息
		  2).消费消息和生产消息并存，这个是事务场景中最常⽤的情况，就是我们常说的consumetransform-
produce 模式
          3).只有consumer消费消息，这种操作其实没有什么意义，跟使用手动提交效果一样，而且也不是事
务属性引入的目的，所以一般不会使用这种情况
   二、几个关键概念和推导
   1).因为producer发送消息可能是分布式事务，所以引入了常用的2PC，所以有事务协调者(Transaction
Coordinator)。Transaction Coordinator和之前为了解决脑裂和惊群问题引入的Group Coordinator在选举
上类似。
   2).事务管理中事务日志是必不可少的，kafka使⽤⼀个内部topic来保存事务⽇志，这个设计和之前使用内部
topic保存偏移量的设计保持⼀致。事务日志是Transaction Coordinator管理的状态的持久化，因为不需要
回溯事务的历史状态，所以事务⽇志只⽤保存最近的事务状态。__transaction_state
   3).因为事务存在commit和abort两种操作，而客户端又有read committed和read uncommitted两种隔离级
别，所以消息队列必须能标识事务状态，这个被称作Control Message。
   4).producer挂掉重启或者漂移到其它机器需要能关联的之前的未完成事务所以需要有一个唯一标识符来进行关
联，这个就是TransactionalId，一个producer挂了，另⼀个有相同TransactionalId的producer能够接着处
理这个事务未完成的状态。kafka目前没有引入全局序，所以也没有transaction id，这个TransactionalId是
用户提前配置的。
   5).TransactionalId能关联producer，也需要避免两个使用相同TransactionalId的producer同时存在，所以引
入了producer epoch来保证对应一个TransactionalId只有一个活跃的producer
   三、事务语义
   1).多分区原子写入
   事务能够保证Kafka topic下每个分区的原子写入。事务中所有的消息都将被成功写入或者丢弃。
   首先，我们来考虑一下原子读取-处理-写入周期是什么意思。简而言之，这意味着如果某个应用程序在某个topic
tp0的偏移量X处读取到了消息A，并且在对消息A进行了一些处理（如B = F（A））之后将消息B写入topic tp1，则只
有当消息A和B被认为被成功地消费并⼀起发布，或者完全不发布时，整个读取过程写入操作是原子的。
   现在，只有当消息A的偏移量X被标记为已消费，消息A才从topic tp0消费，消费到的数据偏移量（record
offset）将被标记为提交偏移量（Committing offset）。在Kafka中，我们通过写入一个名为offsets topic的内部
Kafka topic来记录offset commit。消息仅在其offset被提交给offsets topic时才被认为成功消费。
   由于offset commit只是对Kafkatopic的另一次写入，并且由于消息仅在提交偏移量时被视为成功消费，所以跨多
个主题和分区的原子写入也启用原子读取-处理-写入循环：提交偏移量X到offset topic和消息B到tp1的写入将是单个
事务的一部分，所以整个步骤都是原子的。
   2).粉碎“僵尸实例”
   我们通过为每个事务Producer分配一个称为transactional.id的唯一标识符来解决僵尸实例的问题。在进程重新启
动时能够识别相同的Producer实例。
   API要求事务性Producer的第一个操作应该是在Kafka集群中显示注册transactional.id。 当注册的时候，Kafka
broker用给定的transactional.id检查打开的事务并且完成处理。 Kafka也增加了一个与transactional.id相关的
epoch。Epoch存储每个transactional.id内部元数据。
   一旦epoch被触发，任何具有相同的transactional.id和旧的epoch的生产者被视为僵尸，Kafka拒绝来自这些生产
者的后续事务性写入。
   简而言之：Kafka可以保证Consumer最终只能消费非事务性消息或已提交事务性消息。它将保留来自未完成事务
的消息，并过滤掉已中止事务的消息。
   3).事务消息定义
   生产者可以显式地发起事务会话，在这些会话中发送（事务）消息，并提交或中止事务。有如下要求：
   (1).原子性：消费者的应用程序不应暴露于未提交事务的消息中。
   (2).持久性：Broker不能丢失任何已提交的事务。
   (3).排序：事务消费者应在每个分区中以原始顺序查看事务消息。
   (4).交织：每个分区都应该能够接收来自事务性生产者和非事务生产者的消息
   (5).事务中不应有重复的消息。
   如果允许事务性和非事务性消息的交织，则非事务性和事务性消息的相对顺序将基于附加（对于非事务性消息）
和最终提交（对于事务性消息）的相对顺序。
   在上图中，分区p0和p1接收事务X1和X2的消息，以及非事务性消息。时间线是消息到达Broker的时间。由于首
先提交了X2，所以每个分区都将在X1之前公开来⾃X2的消息。由于非事务性消息在X1和X2的提交之前到达，因此这些
消息将在来自任一事务的消息之前公开。
   四、事务配置
   (1).创建消费者代码，需要：
   将配置中的自动提交属性（auto.commit）进行关闭
   而且在代码里面也不能使用手动提交commitSync( )或者commitAsync( )
   设置isolation.level：READ_COMMITTED或READ_UNCOMMITTED
   (2).创建生成者，代码如下,需要：
   配置transactional.id属性
   配置enable.idempotence属性
   五、事务概览
   生产者将表示事务开始/结束/中止状态的事务控制消息发送给使用多阶段协议管理事务的高可用事务协调器。生产
者将事务控制记录（开始/结束/中止）发送到事务协调器，并将事务的消息直接发送到目标数据分区。消费者需要了解
事务并缓冲每个待处理的事务，直到它们到达其相应的结束（提交/中止）记录为止。
   事务组
   事务组中的生产者
   事务组的事务协调器
   Leader brokers（事务数据所在分区的Broker）
   事务的消费者
   六、事务组
   事务组用于映射到特定的事务协调器（基于日志分区数字的哈希）。该组中的生产者需要配置为该组事务生产
者。由于来自这些生产者的所有事务都通过此协调器进行，因此我们可以在这些事务生产者之间实现严格的有序。
   七、生产者ID和事务组状态
   事务生产者需要两个新参数：生产者ID和生产组。
   需要将生产者的输入状态与上一个已提交的事务相关联。这使事务生产者能够重试事务（通过为该事务重新创建
输入状态；在我们的用例中通常是偏移量的向量）。
   可以使用消费者偏移量管理机制来管理这些状态。消费者偏移量管理器将每个键（ consumergroup-topicpartition
）与该分区的最后一个检查点偏移量和元数据相关联。在事务生产者中，我们保存消费者的偏移量，该偏
移量与事务的提交点关联。此偏移提交记录（在__consumer_offsets 主题中）应作为事务的一部分写入。即，存储
消费组偏移量的__consumer_offsets 主题分区将需要参与事务。因此，假定生产者在事务中间失败（事务协调器随
后到期）；当生产者恢复时，它可以发出偏移量获取请求，以恢复与最后提交的事务相关联的输入偏移量，并从该点
恢复事务处理。
   为了支持此功能，我们需要对偏移量管理器和压缩的__consumer_offsets 主题进行一些增强。
   首先，压缩的主题现在还将包含事务控制记录。我们将需要为这些控制记录提出剔除策略。
   八、事务协调器
   事务协调器是__transaction_state 主题特定分区的Leader分区所在的Broker。它负责初始化、提交以及
回滚事务。事务协调器在内存管理如下的状态：
             对应正在处理的事务的第一个消息的HW。事务协调器周期性地将HW写到ZK。
             事务控制日志中存储对应于日志HW的所有正在处理的事务：
             事务消息主题分区的列表。
			    事务的超时时间。
                与事务关联的Producer ID。
   需要确保无论是什么样的保留策略（日志分区的删除还是压缩），都不能删除包含事务HW的日志分段。
   九、事务流程
   初始阶段
   (图中步骤A)
   (1).Producer：计算哪个Broker作为事务协调器。
   (2).Producer：向事务协调器发送BeginTransaction(producerId, generation, partitions... )请求，当然也可以
发送另一个包含事务过期时间的。如果生产者需要将消费者状态作为事务的一部分提交事务，则需要在
BeginTransaction中包含对应的__consumer_offsets 主题分区信息。
   (3).Broker：生成事务ID
   (4).Coordinator：向事务协调主题追加BEGIN(TxId, producerId, generation, partitions...)消息，然后发送响应
给⽣产者。
   (5).Producer：读取响应（包含了事务ID：TxId）
   (6).Coordinator (and followers)：在内存更新当前事务的待确认事务状态和数据分区信息。
   发送阶段
   (图中步骤2)
   Producer：发送事务消息给主题Leader分区所在的Broker。每个消息需要包含TxId和TxCtl字段。
   TxCtl仅用于标记事务的最终状态（提交还是中止）。生产者请求也封装了生产者ID，但是不追加到日志中。
   结束阶段 (生产者准备提交事务)
   (图中步骤3、4、5。)
   (1).Producer：发送OffsetCommitRequest请求提交与事务结束状态关联的输入状态（如下一个事务输入从哪
儿开始）
   (2)Producer：发送CommitTransaction(TxId, producerId, generation)请求给事务协调器并等待响应。(如果
响应中没有错误信息，表示将提交事务)
   (3).Coordinator：向事务控制主题追加PREPARE_COMMIT(TxId)请求并向生产者发送响应。
   (4).Coordinator：向事务涉及到的每个Leader分区（事务的业务数据的⽬标主题）的Broker发送一个
CommitTransaction(TxId, partitions...)请求。
   (5).事务业务数据的目标主题相关Leader分区Broker：
            1).如果是非__consumer_offsets 主题的Leader分区：一收到CommitTransaction(TxId,
partition1, partition2, ...)请求就会向对应的分区Broker发送空(null)消息(没有key/value)并给该消
息设置TxId和TxCtl（设置为COMMITTED）字段。Leader分区的Broker给协调器发送响应。
            2).如果是__consumer_offsets 主题的Leader分区：追加消息，该消息的key是G-LAST-COMMIT ，
value就是TxId 的值。同时也应该给该消息设置TxId和TxCtl字段。Broker向协调器发送响应。
   (6).Coordinator：向事务控制主题发送COMMITTED(TxId)请求。__transaction_state
   (7).Coordinator (and followers)：尝试更新HW。
   十、事务的中止
   当事务生产者发送业务消息的时候如果发生异常，可以中止该事务。如果事务提交超时，事务协调器也会中止当
前事务。
   Producer：向事务协调器发送AbortTransaction(TxId)请求并等待响应。(一个没有异常的响应表示事务将会
中止)
   Coordinator：向事务控制主题追加PREPARE_ABORT(TxId)消息，然后向生产者发送响应。
   Coordinator：向事务业务数据的目标主题的每个涉及到的Leader分区Broker发送AbortTransaction(TxId,
partitions...)请求。(收到Leader分区Broker响应后，事务协调器中止动作跟上面的提交类似。)
   十一、基本事务流程的失败
   生产者发送BeginTransaction(TxId):的时候超时或响应中包含异常，生产者使用相同的TxId重试。
   生产者发送数据时的Broker错误：生产者应中止（然后重做）事务（使用新的TxId）。如果生产者没有中止
事务，则协调器将在事务超时后中止事务。仅在可能已将请求数据附加并复制到Follower的错误的情况下才
需要重做事务。例如，生产者请求超时将需要重做，而NotLeaderForPartitionException不需要重做。
   生产者发送CommitTransaction(TxId)请求超时或响应中包含异常，生产者使用相同的TxId重试事务。此时
需要幂等性。
   十二、主题的压缩
   压缩主题在压缩过程中会丢弃具有相同键的早期记录。如果这些记录是事务的一部分，这合法吗？ 这可能有点怪
异，但可能不会太有害，因为在主题中使用压缩策略的理由是保留关键数据的最新更新。
   如果该应用程序正在（例如）更新某些表，并且事务中的消息对应于不同的键，则这种情况可能导致数据库视图
不一致。
   十三、事务相关配置
   1)、Broker configs
   配置项                                      说明
   transactional.id.timeout.ms                 在ms中，事务协调器在生产者TransactionalId提前过期之前等待的最长时间，并且没有从该
                                               生产者TransactionalId接收到任何事务状态更新。默认是604800000(7天)。这允许每周一次
                                               的生产者作业维护它们的id        
   max.transaction.timeout.ms                  事务允许的最大超时。如果客户端请求的事务时间超过此时间，broke将在InitPidRequest中
                                               返回InvalidTransactionTimeout错误。这可以防止客户机超时过大，从而导致用户无法从事
                                               务中包含的主题读取内容。
											   默认值为900000(15分钟)。这是消息事务需要发送的时间的保守上限。
   transaction.state.log.replication.factor    事务状态topic的副本数量。默认值:3
   transaction.state.log.num.partitions        事务状态主题的分区数。默认值:50
   transaction.state.log.min.isr               事务状态主题的每个分区ISR最小数量。默认值:2
   transaction.state.log.segment.bytes         事务状态主题的segment大小。默认值:104857600 字节
   2)、Producer configs
   配置项                  说明 
   enable.idempotence      开启幂等
   transaction.timeout.ms  事务超时时间
                           事务协调器在主动中止正在进行的事务之前等待生产者更新事务状态的最长时
                           间。这个配置值将与InitPidRequest一起发送到事务协调器。如果该值大于
                           max.transaction.timeout。在broke中设置ms时，请求将失败，并出现
                           InvalidTransactionTimeout错误。
						   默认是60000。这使得交易不会阻塞下游消费超过一分钟，这在实时应用程序中
                           通常是允许的。
   transactional.id        用于事务性交付的TransactionalId。这支持跨多个生产者会话的可靠性语义，
                           因为它允许客户端确保使用相同TransactionalId的事务在启动任何新事务之前
                           已经完成。如果没有提供TransactionalId，则生产者仅限于幂等交付。
   3)、Consumer configs
   配置项            说明
   isolation.level  - read_uncommitted:以偏移顺序使用已提交和未提交的消息。
                    - read_committed:仅以偏移量顺序使用非事务性消息或已提交事务性消息。为了维护偏
                      移排序，这个设置意味着我们必须在使用者中缓冲消息，直到看到给定事务中的所有消
                      息。
   
  